package com.bff.gaia.mix.api.xjoin.client.rdb.async;
/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import io.vertx.core.json.JsonArray;
import io.vertx.ext.sql.SQLClient;
import io.vertx.ext.sql.SQLConnection;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.functions.async.ResultFuture;
import com.bff.gaia.mix.api.xjoin.core.enums.ECacheContentType;
import com.bff.gaia.mix.api.xjoin.core.side.AsyncReqRow;
import com.bff.gaia.mix.api.xjoin.core.side.CacheMissVal;
import com.bff.gaia.mix.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.mix.api.xjoin.core.side.JoinInfo;
import com.bff.gaia.mix.api.xjoin.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin.client.rdb.utils.SwitchUtil;
import com.bff.gaia.mix.api.xjoin.core.side.cache.CacheObj;
import com.bff.gaia.shaded.guava18.com.google.common.collect.Lists;
import com.bff.gaia.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.util.List;
import java.util.Map;

/**
 * Project Name: gaia-parent
 * Description:
 * Data: 2019/7/19 16:38
 *
 * @author tiger
 * @version v1.0
 */
public class RdbAsyncReqRow extends AsyncReqRow {

	private static final long serialVersionUID = 2098635244857937720L;

	private static final Logger LOG = LoggerFactory.getLogger(RdbAsyncReqRow.class);

	public final static int DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE = 1;

	public final static int DEFAULT_VERTX_WORKER_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;

	public final static int DEFAULT_MAX_DB_CONN_POOL_SIZE = DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE + DEFAULT_VERTX_WORKER_POOL_SIZE;

	private transient SQLClient rdbSQLClient;

	public RdbAsyncReqRow(RowTypeInfo rowTypeInfo, JoinInfo joinInfo, List<FieldInfo> outFieldInfoList, SideTableInfo sideTableInfo) {
		super(new RdbAsyncSideInfo(rowTypeInfo, joinInfo, outFieldInfoList, sideTableInfo));
	}

	@Override
	public void asyncInvoke(Row input, ResultFuture<Row> resultFuture) throws Exception {
//		System.out.println(input);
		JsonArray inputParams = new JsonArray();
		for (Integer conValIndex : sideInfo.getEqualValIndex()) {
			Object equalObj = input.getField(conValIndex);
//			System.out.println(conValIndex+" "+equalObj);
			if (equalObj == null) {
				resultFuture.complete(null);
				return;
			}
			inputParams.add(equalObj);
		}

		String key = buildCacheKey(inputParams);
//		System.out.println(key);
		if (openCache()) {
			CacheObj val = getFromCache(key);
			if (val != null) {

				if (ECacheContentType.MissVal == val.getType()) {
					dealMissKey(input, resultFuture);
					return;
				} else if (ECacheContentType.MultiLine == val.getType()) {
					List<Row> rowList = Lists.newArrayList();
					for (Object jsonArray : (List) val.getContent()) {
						Row row = fillData(input, jsonArray);
						rowList.add(row);
					}
					resultFuture.complete(rowList);
				} else {
					throw new RuntimeException("not support cache obj type " + val.getType());
				}
				return;
			}
		}

		rdbSQLClient.getConnection(conn -> {
			if (conn.failed()) {
				//Treatment failures
				resultFuture.completeExceptionally(conn.cause());
				return;
			}

			final SQLConnection connection = conn.result();
			String sqlCondition = sideInfo.getSqlCondition();
			connection.queryWithParams(sqlCondition, inputParams, rs -> {
				if (rs.failed()) {
					LOG.error("Cannot retrieve the data from the database", rs.cause());
					resultFuture.complete(null);
					return;
				}

				List<JsonArray> cacheContent = Lists.newArrayList();

				int resultSize = rs.result().getResults().size();
				if (resultSize > 0) {
					List<Row> rowList = Lists.newArrayList();

					for (JsonArray line : rs.result().getResults()) {
//						//TODO
//						System.out.println(input);
//						System.out.println(line);
						Row row = fillData(input, line);
						if (openCache()) {
							cacheContent.add(line);
						}
						rowList.add(row);
					}

					if (openCache()) {
						putCache(key, CacheObj.buildCacheObj(ECacheContentType.MultiLine, cacheContent));
					}

					resultFuture.complete(rowList);
				} else {
					dealMissKey(input, resultFuture);
					if (openCache()) {
						putCache(key, CacheMissVal.getMissKeyObj());
					}
				}

				// and close the connection
				connection.close(done -> {
					if (done.failed()) {
						throw new RuntimeException(done.cause());
					}
				});
			});
		});
	}

	@Override
	public Row fillData(Row input, Object line) {
		JsonArray jsonArray = (JsonArray) line;
		Row row = new Row(sideInfo.getOutFieldInfoList().size());
		String[] fields = sideInfo.getSideTableInfo().getFieldTypes();
		for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
//			System.out.println(input+" "+entry.getValue());
			Object obj = input.getField(entry.getValue());
//			boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(xjoinInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
			boolean isTimeIndicatorTypeInfo = false;
			if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
				obj = ((Timestamp) obj).getTime();
			}

			row.setField(entry.getKey(), obj);
		}

		int i=0;
		for (Map.Entry<Integer, Integer> entry : sideInfo.getSideFieldIndex().entrySet()) {
			if (jsonArray == null) {
				row.setField(entry.getKey(), null);
			} else {
				Object object = SwitchUtil.getTarget(jsonArray.getValue(i), fields[entry.getValue()]);
				row.setField(entry.getKey(), object);
				i++;
			}
		}
		return  Row.project(row, sideInfo.getOutFieldIndexs());
	}

	@Override
	public void close() throws Exception {
		super.close();
		if (rdbSQLClient != null) {
			rdbSQLClient.close();
		}

	}

	public String buildCacheKey(JsonArray jsonArray) {
		StringBuilder sb = new StringBuilder();
		for (Object ele : jsonArray.getList()) {
			sb.append(ele.toString())
				.append("_");
		}

		return sb.toString();
	}

	public void setRdbSQLClient(SQLClient rdbSQLClient) {
		this.rdbSQLClient = rdbSQLClient;
	}
}